package com.hujf.flink.panglu;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author hujf
 * @Date 2022/5/24 11:09
 * @Description TODO
 */
public class SideOutPutDemo {
    public static void main(String[] args) {
        final OutputTag<String> outputTag = new OutputTag<String>("side1");
        final OutputTag<String> outputTag2 = new OutputTag<String>("side2");

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream <Integer> input = environment.fromElements(1,2,3,4);

        SingleOutputStreamOperator mainStream = input.process(new ProcessFunction<Integer, Integer>() {
            @Override
            public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
                out.collect(value);
                ctx.output(outputTag,"side-"+String.valueOf(value));
                ctx.output(outputTag2,"side2-"+String.valueOf(value*3));
            }
        });
    }
}
